/////////////////////////////////////////////////////////////////////////////
// Original code from libhdfs3. Copyright (c) 2013 - 2014, Pivotal Inc.
// All rights reserved. Author: Zhanwei Wang
/////////////////////////////////////////////////////////////////////////////
//  Modifications by Kumo Inc.
// Copyright (C) Kumo inc. and its affiliates.
// Author: Jeff.li lijippy@163.com
// All rights reserved.
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published
// by the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program.  If not, see <https://www.gnu.org/licenses/>.
//



#include <turbo/times/time.h>
#include <kmhdfs/common/exception.h>
#include <kmhdfs/common/exception_internal.h>
#include <kmhdfs/client/file_system_inter.h>
#include <kmhdfs/client/lease_renewer.h>

#define DEFAULT_LEASE_RENEW_INTERVAL (60 * 1000)

namespace Hdfs {
namespace Internal {

once_flag LeaseRenewer::once;
std::shared_ptr<LeaseRenewer> LeaseRenewer::renewer;

LeaseRenewer & LeaseRenewer::GetLeaseRenewer() {
    call_once(once, &LeaseRenewer::CreateSinglten);
    assert(renewer);
    return *renewer;
}

void LeaseRenewer::CreateSinglten() {
    renewer = std::shared_ptr < LeaseRenewer > (new LeaseRenewerImpl());
}

LeaseRenewerImpl::LeaseRenewerImpl() :
    stop(true), interval(DEFAULT_LEASE_RENEW_INTERVAL) {
}

LeaseRenewerImpl::~LeaseRenewerImpl() {
    stop = true;
    cond.notify_all();

    if (worker.joinable()) {
        worker.join();
    }
}

int LeaseRenewerImpl::getInterval() const {
    return interval;
}

void LeaseRenewerImpl::setInterval(int interval) {
    this->interval = interval;
}

void LeaseRenewerImpl::StartRenew(std::shared_ptr<FileSystemInter> filesystem) {
    lock_guard<mutex> lock(mut);
    const char * clientName = filesystem->getClientName();

    if (maps.find(clientName) == maps.end()) {
        maps[clientName] = filesystem;
    }

    filesystem->registerOpenedOutputStream();

    if (stop && !maps.empty()) {
        if (worker.joinable()) {
            worker.join();
        }

        stop = false;
        CREATE_THREAD(worker, std::bind(&LeaseRenewerImpl::renewer, this));
    }
}

void LeaseRenewerImpl::StopRenew(std::shared_ptr<FileSystemInter> filesystem) {
    lock_guard<mutex> lock(mut);
    const char * clientName = filesystem->getClientName();

    if (filesystem->unregisterOpenedOutputStream()
            && maps.find(clientName) != maps.end()) {
        maps.erase(clientName);
    }
}

void LeaseRenewerImpl::renewer() {
    assert(stop == false);

    while (!stop) {
        try {
            unique_lock < mutex > lock(mut);
            cond.wait_for(lock, std::chrono::milliseconds(interval));

            if (stop || maps.empty()) {
                break;
            }

            std::map<std::string, std::shared_ptr<FileSystemInter> >::iterator s, e, d;
            e = maps.end();

            for (s = maps.begin(); s != e;) {
                std::shared_ptr<FileSystemInter> fs = s->second;

                try {
                    if (!fs->renewLease()) {
                        d = s++;
                        maps.erase(d);
                    } else {
                        ++s;
                    }

                    continue;
                } catch (const HdfsException & e) {
                    std::string buffer;
                    KLOG(ERROR)<<turbo::str_format(
                        "Failed to renew lease for filesystem which client name is %s, since:\n%s",
                        fs->getClientName(), GetExceptionDetail(e, buffer));
                } catch (const std::exception & e) {
                    KLOG(ERROR)<<turbo::str_format(
                        "Failed to renew lease for filesystem which client name is %s, since:\n%s",
                        fs->getClientName(), e.what());
                    break;
                }

                ++s;
            }

            if (maps.empty()) {
                break;
            }
        } catch (const std::bad_alloc & e) {
            /*
             * keep quiet if we run out of memory, since writing log needs memory,
             * that may cause the process terminated.
             */
            break;
        } catch (const std::exception & e) {
            KLOG(ERROR)<<turbo::str_format(
                "Lease renewer will exit since unexpected exception: %s",
                e.what());
            break;
        }
    }

    stop = true;
}

}
}
